package com.atguigu.gmall.common.config;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.common.model.GmallCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @Description 消息发送确认-确保生产者消息不丢失
 * <p>
 * ConfirmCallback  只确认消息是否正确到达 Exchange 中
 * ReturnCallback   消息没有正确到达队列时触发回调，如果正确到达队列不执行
 * <p>
 * 1.如果消息没有到exchange,则confirm回调,ack=false
 * 2.如果消息到达exchange,则confirm回调,ack=true
 * 3.exchange到queue成功,则不回调return
 * 4.exchange到queue失败,则回调return
 */
@Slf4j
@Component
public class MQProducerAckConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 应用启动后触发一次
     */
    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    /**
     * 交换机确认:当消息发送到交换机,交换机将结果(成功或者失败)异步发送给生成者
     * @param correlationData 回调的相关性数据
     * @param ack             应答结果 true:发送成功,false:失败
     * @param cause           如果发送消息到交换器失败，错误原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("{生产者确认之交换机确认},相关数据:{},回调结果:{},失败原因:{}",correlationData,ack,cause);
        if (ack) {
            //消息到交换器成功
            log.info("消息发送到Exchange成功：{}", correlationData);
        } else {
            //消息到交换器失败
            log.error("消息发送到Exchange失败：{}", cause);
            //todo 消息重发
            GmallCorrelationData gmallCorrelationData = (GmallCorrelationData) correlationData;
            this.retrySendMsg(gmallCorrelationData);
        }
    }

    /**
     * 消息重发
     * @param gmallCorrelationData
     */
    private void retrySendMsg(GmallCorrelationData gmallCorrelationData) {
        //获取相关数据:判断是否满足重试条件;进行重发消息
        String uuid = gmallCorrelationData.getId();
        int retryCount = gmallCorrelationData.getRetryCount();
        if (retryCount >= 3){
            log.error("重试达到上限;将失败的消息写入死信队列;有单独消费者读取死信队列中消息;人工处理");
            return;
        }
        //调用模拟对象将消息重新发送一次
        System.out.println("进行消息重发");
        gmallCorrelationData.setRetryCount(gmallCorrelationData.getRetryCount() + 1);
        rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(),gmallCorrelationData.getRoutingKey(),gmallCorrelationData.getMessage(),gmallCorrelationData);
        redisTemplate.opsForValue().set(uuid, JSON.toJSONString(gmallCorrelationData));
        rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(),gmallCorrelationData.getRoutingKey(),gmallCorrelationData.getMessage(),gmallCorrelationData);
    }


    /**
     * 交换机确认之队列确认:交换机路由消息队列失败才会发送给生成者
     * @param message 返回消息对象
     * @param replyCode 应答码
     * @param replyText 错误的原因
     * @param exchange 交换机的名称
     * @param routingKey 路由键
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        //对消息进行重发
        String redisKey = message.getMessageProperties().getHeader("spring_returned_message_correlation");
        String str = (String) redisTemplate.opsForValue().get(redisKey);
        GmallCorrelationData gmallCorrelationData = JSON.parseObject(str, GmallCorrelationData.class);
        //交换机暂存消息,消息无法及时路由到队列,会触发队列路由异常回调方法
        //方式一:如果是延迟消息,不处理
        if (gmallCorrelationData.isDelay()){
            return;
        }
        //方式二:对重发方法进行修正,增加延迟消息重发逻辑,必然导致延迟消息多发
        log.error("消息路由queue失败，应答码={}，原因={}，交换机={}，路由键={}，消息={}",
                replyCode, replyText, exchange, routingKey, message.toString());
        this.retryMessage(gmallCorrelationData);
    }

    /**
     * 消息重发
     * @param correlationData 相关数据
     */
    private void retryMessage(GmallCorrelationData correlationData) {
        //获取重试次数
        int retryCount = correlationData.getRetryCount();
        //判断重试次数>=3
        if (retryCount >= 3){
            log.error("重发次数已达上限,交给运维人员处理,将发送失败消息存入异常数据库,人工处理!");
            return;
        }
        //更新Redis中重试次数
        String correlationDataStr = (String) redisTemplate.opsForValue().get(correlationData.getId());
        GmallCorrelationData gmallCorrelationData = JSON.parseObject(correlationDataStr, GmallCorrelationData.class);
        gmallCorrelationData.setRetryCount(gmallCorrelationData.getRetryCount() + 1);
        redisTemplate.opsForValue().set(correlationData.getId(),JSON.toJSONString(gmallCorrelationData));
        //处理延迟消息重发,保证延迟消息依然发送
        if (gmallCorrelationData.isDelay()){
            //结果:对于延迟性消息立即执行重发方法
            rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(),gmallCorrelationData.getRoutingKey(),gmallCorrelationData.getMessage(),message -> {
                message.getMessageProperties().setDelay(gmallCorrelationData.getDelayTime() * 1000);
                return message;
            },gmallCorrelationData);
        }else {
            //重相关数据中获取所有信息用于重发消息
            rabbitTemplate.convertAndSend(correlationData.getExchange(),correlationData.getRoutingKey(),correlationData.getMessage(),gmallCorrelationData);
        }
        System.out.println("---------消息进行重发------------");
    }
}